package demo.base.connect;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.RichCoFlatMapFunction;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

public class BasicCoFlatMapFunc<IN1 extends Order, IN2 extends Order, OUT extends Order>
        extends RichCoFlatMapFunction<IN1, IN2, OUT> {
    Class<? extends Order> classIn1 = null;
    Class<? extends Order> classIn2 = null;
    Class<? extends Order> classOut = null;
    ValueState<IN1> in1VState;
    ValueState<IN2> in2VState;

    @Override
    public void open(Configuration parameters) throws Exception {
        getClassMethod();

        in1VState =
                getRuntimeContext()
                        .getState(
                                new ValueStateDescriptor<IN1>("in1VState", (Class<IN1>) classIn1));
        in2VState =
                getRuntimeContext()
                        .getState(
                                new ValueStateDescriptor<IN2>("in2VState", (Class<IN2>) classIn2));
    }

    @Override
    public void flatMap1(IN1 value, Collector<OUT> out) throws Exception {
        String orderId = value.getOrderId();
        IN2 tmpIn1 = in2VState.value();
        IN2 curIn2 = tmpIn1 == null ? (IN2) value.coInit(orderId) : tmpIn1;

        in1VState.update(value);

        System.out.println("flatmap1: " + JSON.toJSONString(value));
        System.out.println("flatmap1 curIn2: " + JSON.toJSONString(curIn2));
        //        out.collect((OUT) classOut.newInstance().collect(JSON.toJSONString(value),
        // JSON.toJSONString(curIn2)));
        Method collect = classOut.getMethod("collect", String.class, String.class);
        Order order = classOut.newInstance();
        Object invoke = collect.invoke(order, JSON.toJSONString(value), JSON.toJSONString(curIn2));
        out.collect((OUT) invoke);
    }

    @Override
    public void flatMap2(IN2 value, Collector<OUT> out) throws Exception {
        String orderId = value.getOrderId();
        IN1 tmpIn2 = in1VState.value();
        IN1 curIn2 = tmpIn2 == null ? (IN1) value.coInit(orderId) : tmpIn2;

        in2VState.update(value);

        System.out.println("flatMap2: " + JSON.toJSONString(value));
        System.out.println("flatMap2 curIn2: " + JSON.toJSONString(curIn2));
        //        out.collect((OUT) classOut.newInstance().collect(JSON.toJSONString(curIn2),
        // JSON.toJSONString(value)));
        Method collect = classOut.getMethod("collect", String.class, String.class);
        Order order = classOut.newInstance();
        Object invoke = collect.invoke(order, JSON.toJSONString(curIn2), JSON.toJSONString(value));
        out.collect((OUT) invoke);
    }

    public void getClassMethod() {
        Type superclass = getClass().getGenericSuperclass();
        ParameterizedType parameterizedType = null;
        if (superclass instanceof ParameterizedType) {
            parameterizedType = (ParameterizedType) superclass;
            Type[] typeArray = parameterizedType.getActualTypeArguments();
            if (typeArray != null && typeArray.length > 0) {
                classIn1 = (Class<? extends Order>) typeArray[0];
                classIn2 = (Class<? extends Order>) typeArray[1];
                classOut = (Class<? extends Order>) typeArray[2];
                //                System.out.println("======" + classIn1 + "======");
                //                System.out.println("======" + classIn2 + "======");
                //                System.out.println("======" + classOut + "======");
            }

        } else {
            System.out.println("##NOT##");
        }
    }
}
